package com.test.study.ToolsUtils.strom.trident;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FileUtils;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.spout.IBatchSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class MyFixedBatchSpout implements IBatchSpout {

	private static final long serialVersionUID = 4161477428435032380L;
	HashMap<Long, List<List<Object>>> batches = new HashMap<Long, List<List<Object>>>();

	@SuppressWarnings("rawtypes")
	@Override
	public void open(Map conf, TopologyContext context) {
		// TODO Auto-generated method stub

	}

	@SuppressWarnings("deprecation")
	@Override
	public void emitBatch(long batchId, TridentCollector collector) {
		List<List<Object>> batch = this.batches.get(batchId);
		if (batch == null) {
			batch = new ArrayList<List<Object>>();
			Collection<File> listFiles = FileUtils.listFiles(new File("/Users/huyong/App/file/testDir"),
					new String[] { "txt" }, true);
			if (listFiles != null && listFiles.size() > 0) {
				for (File file : listFiles) {
					try {
						List<String> readLines = FileUtils.readLines(file);
						for (String line : readLines) {
							batch.add(new Values(line));
						}
					} catch (IOException e) {
						e.printStackTrace();
					}
					try {
						FileUtils.moveFile(file, new File(file.getAbsolutePath() + System.currentTimeMillis()));
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
				this.batches.put(batchId, batch);
			}
			for (List<Object> list : batch) {
				collector.emit(list);
			}
		}

	}

	@Override
	public void ack(long batchId) {

	}

	@Override
	public void close() {
		// TODO Auto-generated method stub

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Fields getOutputFields() {

		return new Fields("line");
	}

}
